package com.dmall.scf

import com.dmall.scf.utils.{MySQLUtils, SparkUtils}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
 * @descrption 定义Spark处理流程
 * @author wangxuexing
 * @date 2019/12/23
 */
trait SparkAction[T] {
  /**
   * 定义流程
   */
  def execute(args: Array[String], spark: SparkSession)={
    //1. 前置处理
    preAction
    //2. 处理
    val df = action(spark, args)
    //3. 后置处理
    postAction(df)
  }

  /**
   * 前置处理
   * @return
   */
  def preAction() = {
    //无前置处理
  }

  /**
   * 处理
   * @param spark
   * @return
   */
  def action(spark: SparkSession, args: Array[String]) : DataFrame

  /**
   * 后置处理，比如保存结果到Mysql
   * @param df
   */
  def postAction(df: DataFrame)={
    //结果追加到scfc_supplier_run_field_value表
    MySQLUtils.writeIntoMySql(df, saveTable._1, saveTable._2)
  }

  /**
   * 保存mode及表名
   * @return
   */
  def saveTable: (SaveMode, String)
}
